package cn.com.fattiger.framework.remoting.netty4.client;

import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantLock;

import cn.com.fattiger.framework.remoting.netty4.client.channel.GlobalClientChannelManager;
import cn.com.fattiger.framework.remoting.netty4.common.DefaultCallback;
import cn.com.fattiger.framework.remoting.netty4.common.ICallback;
import cn.com.fattiger.framework.remoting.netty4.common.Invocation;
import cn.com.fattiger.framework.remoting.netty4.common.channel.ChannelManager;
import cn.com.fattiger.framework.remoting.netty4.common.protocol.params.Constants;

/**
 * 
 * @author fattiger.xiaoyang
 * @date 2016/01/26
 * @qqGroup 48450434
 */
public class ResponseFuture {

	/* UUID --> channel 对应关系，用于重新发送 */
	private static final Map<Long, ChannelManager> channelManagerMaps = new ConcurrentHashMap<Long, ChannelManager>();

	/* 鉴于通道信息无归属行,需要维护所述关系 */
	private static final Map<Long, ResponseFuture> futureMaps = new ConcurrentHashMap<Long, ResponseFuture>();

	/* 请求唯一编号 */
	private final long id;

	/* channel 管理 */
	private final ChannelManager ChannelManager;

	/* 请求内容 */
	private final Invocation request;

	/* 超时时间 */
	private final int timeout;

	/* 可重入锁 : 进行一步返回唤醒阻塞获取结果县城 */
	private final Lock lock = new ReentrantLock();
	private final Condition done = lock.newCondition();

	/* 超时时间 */
	private final long start = System.currentTimeMillis();

	private volatile String response;

	/* 重试次数 */
	private volatile int timeOutCount = 0;// 超时次数

	/* 异步回调 */
	private ICallback callback;

	/* 超时扫描线程 */
	static {
		Thread th = new Thread(new RemotingInvocationTimeoutScan(), "ResponseTimeout");
		th.setDaemon(true);
		th.start();
	}

	public ResponseFuture(ChannelManager channelManager, Invocation request, int timeout, ICallback callback) {
		this.ChannelManager = channelManager;
		this.request = request;
		this.id = request.getUuid();
		this.timeout = timeout > 0 ? timeout : Constants.DEFAULT_TIMEOUT;
		this.callback = (null == callback) ? new DefaultCallback(this) : callback;
		futureMaps.put(id, this);
		channelManagerMaps.put(id, channelManager);
	}

	public Object get() throws Exception {
		return get(timeout);
	}

	/**
	 * @param timeout
	 * @return
	 * @throws Exception
	 */
	public Object get(int timeout) throws Exception {
		if (timeout <= 0) {
			timeout = Constants.DEFAULT_TIMEOUT;
		}
		if (!isDone()) {
			lock.lock();
			try {
				while (!isDone()) {
					done.await(timeout, TimeUnit.MILLISECONDS);
					if (isDone()) {
						break;
					}
					if (timeOutCount >= Constants.TimeOutCount) {
						throw new Exception(id + ":超时");
					}
				}
			} catch (InterruptedException e) {
				throw new RuntimeException(e);
			} finally {
				lock.unlock();
			}
		}
		return getResponse();
	}

	private String getResponse() {
		return response;
	}

	public void setResponse(Object obj) {
		this.response = String.valueOf(obj);
	}

	public boolean isDone() {
		return response != null;
	}

	public long getTimeout() {
		return timeout;
	}

	public long getStart() {
		return start;
	}

	public int getTimeOutCount() {
		return timeOutCount;
	}

	public void setTimeOutCount(int timeOutCount) {
		this.timeOutCount = timeOutCount;
	}

	private void doReceived(Invocation res) {
		lock.lock();
		try {
			// 异步回调处理
			if (null != callback) {
				callback.process(res.getResponse());
			}

			if (done != null) {
				done.signal();
			}

		} finally {
			lock.unlock();
		}

	}

	public ChannelManager getChannelManager() {
		return ChannelManager;
	}

	public static void received(Invocation response) {
		try {
			ResponseFuture future = futureMaps.remove(response.getUuid());
			if (future != null) {
				future.doReceived(response);
			} else {
				// System.out.println(response+"：该结果已经执行完毕！");
			}
		} finally {
			ChannelManager channelManager = channelManagerMaps.get(response.getUuid());
			if (channelManager != null) {
				channelManager.setTransferTime(System.currentTimeMillis());
			}
			channelManagerMaps.remove(response.getUuid());
		}
	}

	private static class RemotingInvocationTimeoutScan implements Runnable {

		@Override
		public void run() {
			while (true) {
				try {
					for (ResponseFuture future : futureMaps.values()) {
						if (future == null || future.isDone()) {
							continue;
						}
						if (System.currentTimeMillis() - future.getStart() > future.getTimeout()) {
							if (future.getTimeout() < Constants.TimeOutCount) {
								future.setTimeOutCount(future.getTimeOutCount() + 1);
								ChannelManager channelManager = channelManagerMaps.get(future.id);
								channelManager = GlobalClientChannelManager.get(channelManager.getHostName());
								channelManager.getChannel().writeAndFlush(future.request);
								channelManagerMaps.put(future.id, channelManager);
							} else {
								Invocation timeoutResponse = new Invocation(future.id);
								timeoutResponse.setResponse("获取数据超时");
								ResponseFuture.received(timeoutResponse);
							}
						}
					}
					Thread.sleep(1);
				} catch (Throwable e) {

				}
			}
		}
	}

}
